package cn.itcast.streaming.task;


import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.OnlineDataObj;
import cn.itcast.streaming.bean.VehicleInfoModel;
import cn.itcast.streaming.function.asyncio.AsyncHttpQueryFunction;
import cn.itcast.streaming.function.flatmap.VehicleInfoMysqlFunction;
import cn.itcast.streaming.function.map.LocationInfoRedisFunction;
import cn.itcast.streaming.function.watermark.OnlineStatisticsWatermark;
import cn.itcast.streaming.function.window.OnlineStatisticsWindowFunction;
import cn.itcast.streaming.sink.OnlineStatisticsToMysqlSink;
import cn.itcast.streaming.source.MysqlVehicleInfoFunction;
import cn.itcast.streaming.utils.JsonParsePartUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.util.HashMap;
import java.util.concurrent.TimeUnit;

//在线故障实时分析业务开发
//消费kafka数据进行实时故障分析，将分析结果写入mysql
public class OnlineStatisticsTask extends BaseTask{
    public static void main(String[] args) {
        /**
         * 1.初始化flink环境
         * 2.接入kafka数据源
         * 3.将消费到的json字符串转换成javabean
         * 4.过滤掉异常数据,获取到正常数据
         * 5.与redis维度表数据进行关联拉宽地理位置数据，对拉宽后的流数据关联redis,根据geohash找到地理位置信息
         * 6.过滤出来redis拉宽成功的地理位置数据
         * 7.过滤出来redis拉宽失败的地理位置数据
         * 8.将redis拉宽失败的地理位置数据使用异步io的方式连接高德地图api进行二次拉宽
         * 9.将redis拉宽成功的地理位置数据与高德地图拉宽成功的地理位置数据进行合并
         * 10.创建原始数据30s滚动窗口，根据vin进行分流
         * 11.对原始数据的窗口流数据进行实时故障分析，区分出异常告警的字段和非异常告警的字段
         * 12.加载业务中间表的（车辆类型表，车辆表，车辆用途表，车辆销售记录表）
         * 13.将11和12的广播流进行关联进行拉宽操作
         * 14.将拉宽后的数据写入mysql中
         * 15，地交作业。
         */
        //todo 1.初始化flink环境
        StreamExecutionEnvironment env = getEnv("OnlineStatisticsTask");
        //todo 2.接入kafka数据源
        DataStream<String> kafkaStream = createKafkaStream(SimpleStringSchema.class);
        //todo 3.将消费到的json字符串转换成javabean
        SingleOutputStreamOperator<ItcastDataPartObj> itcastDataPartObjStream = kafkaStream
                .filter(obj -> StringUtils.isNotEmpty(obj))
                .map(JsonParsePartUtil::parseJsonToObject);
        //todo 4.过滤掉异常数据,获取到正常数据
        SingleOutputStreamOperator<ItcastDataPartObj> srcDataPartObjStream = itcastDataPartObjStream
                .filter(obj -> StringUtils.isEmpty(obj.getErrorData()));
        //todo 5.与redis维度表数据进行关联拉宽地理位置数据，(对拉宽后的流数据关联redis,根据geohash找到地理位置信息)
        SingleOutputStreamOperator<ItcastDataPartObj> itcastLocationDataPartObjStream = srcDataPartObjStream
                .map(new LocationInfoRedisFunction());

        //todo 6.过滤出来redis拉宽成功的地理位置数据
        SingleOutputStreamOperator<ItcastDataPartObj> objWithLocationStream = itcastLocationDataPartObjStream
                .filter(itcastDataPartObj -> StringUtils.isNotEmpty(itcastDataPartObj.getProvince()));
        objWithLocationStream.print("使用redis拉宽地理位置数据结果>>>");

        //todo 7.过滤出来redis拉宽失败的地理位置数据
        SingleOutputStreamOperator<ItcastDataPartObj> objNoWithLocationStream = itcastLocationDataPartObjStream
                .filter(itcastDataPartObj -> StringUtils.isEmpty(itcastDataPartObj.getProvince()));
        //todo 8.将redis拉宽失败的地理位置数据使用异步io的方式连接高德地图api进行二次拉宽
        SingleOutputStreamOperator<ItcastDataPartObj> asyncItcastDataPartObjWithLocationStream
                = AsyncDataStream.unorderedWait(
                        objNoWithLocationStream,
                        new AsyncHttpQueryFunction(),
                        2000, TimeUnit.MILLISECONDS,//超过2s为超时
                        10);                                //最大异步并发请求数量（线程队列）
        asyncItcastDataPartObjWithLocationStream.print("使用API拉宽地理位置的数据结果>>");

        //todo 9.将redis拉宽成功的地理位置数据与高德地图拉宽成功的地理位置数据进行合并
        DataStream<ItcastDataPartObj> itcastDataPartObjLocationStream = objWithLocationStream
                .union(asyncItcastDataPartObjWithLocationStream);
        itcastDataPartObjLocationStream.print("地理位置拉宽后的结果数据>>>");
        //todo 10.创建原始数据30s滚动窗口，根据vin进行分流
        WindowedStream<ItcastDataPartObj, String, TimeWindow> itcastWindowDataStream = itcastDataPartObjLocationStream
                .assignTimestampsAndWatermarks(new OnlineStatisticsWatermark())
                .keyBy(obj -> obj.getVin())
                .window(TumblingEventTimeWindows.of(Time.seconds(30)));
        //todo 11.对原始数据的窗口流数据进行实时故障分析，区分出异常告警的字段和非异常告警的字段
        SingleOutputStreamOperator<OnlineDataObj> onlineDataObjStream = itcastWindowDataStream
                .apply(new OnlineStatisticsWindowFunction());
        //todo 12.加载业务中间表的（车辆类型表，车辆表，车辆用途表，车辆销售记录表）
        DataStream<HashMap<String, VehicleInfoModel>> vehicleInfoDataStream = env.addSource(new MysqlVehicleInfoFunction()).broadcast();
        //todo 13.将11和12的广播流进行关联进行拉宽操作
        SingleOutputStreamOperator<OnlineDataObj> connectVehicleInfoStream = onlineDataObjStream
                .connect(vehicleInfoDataStream).flatMap(new VehicleInfoMysqlFunction());
        connectVehicleInfoStream.print("最终拉宽后的数据模型。");
        //todo 14.将拉宽后的数据写入mysql中
        connectVehicleInfoStream.addSink(new OnlineStatisticsToMysqlSink());
        //todo 15，地交作业。
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
